package cn.me.alphamq.common.remote;

import cn.me.alphamq.common.constant.GlobalConstant;
import cn.me.alphamq.common.entity.req.PullBrokerAddressReq;
import cn.me.alphamq.common.entity.req.ServiceHeartBeatReq;
import cn.me.alphamq.common.entity.req.ServiceRegistryReq;
import cn.me.alphamq.common.entity.resp.PullBrokerAddressResp;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.enumeration.ServiceInstanceTypeEnum;
import cn.me.alphamq.common.event.EventBus;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.util.AssertUtil;
import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 默认的客户端，是抽象类，生产者/消费者可以继承它，从而避免写一些重复的代码
 *
 * @author f
 */
public abstract class DefaultClient {

    private static final Logger log = LoggerFactory.getLogger(DefaultClient.class);

    /**
     * 待拉取地址的 broker 的角色
     *
     * @see cn.me.alphamq.common.enumeration.BrokerRoleEnum
     */
    private final String brokerRole;

    /**
     * 待拉取地址的 broker 的组名
     */
    private final String brokerGroup;

    /**
     * 与 nameserver 通信的客户端
     */
    protected final NettyClient nameServerClient;

    /**
     * 建立连接的 broker 的客户端集合
     */
    private volatile Map<String, NettyClient> brokerClientMap = new ConcurrentHashMap<>();

    public DefaultClient(String brokerRole, String brokerGroup,
                         String nameServerIp, int nameServerPort, String nameServerClientName) {
        this.brokerRole = brokerRole;
        this.brokerGroup = brokerGroup;
        this.nameServerClient = new NettyClient(nameServerIp, nameServerPort);
        this.nameServerClient.initConnection(nameServerClientName, new NameServerRespHandler());
    }

    /**
     * 启动客户端，子类可以在构造器中使用
     *
     * @param username 与 NameServer 连接时校验的用户名
     * @param password 与 NameServer 连接时校验的密码
     * @param siType   服务实例的类型
     */
    protected void start(String username, String password, ServiceInstanceTypeEnum siType) {
        registry(username, password, siType);

        startHeartBeatTask();

        connectBroker();
        startRefreshBrokerConnectionTask();

        afterStartCallback();
    }

    // 将本客户端注册到 nameserver 中
    private void registry(String username, String password, ServiceInstanceTypeEnum serviceInstanceType) {
        TcpMsg resp = nameServerClient.sendMsgSync(EventCodeEnum.REGISTRY_REQ, new ServiceRegistryReq(
                username, password, serviceInstanceType.getCode()
        ));
        if (EventCodeEnum.REGISTRY_SUCCESS_RESP.getCode() != resp.getCode()) {
            throw new RuntimeException("registry failed...");
        }
        log.info("registry success...");
    }

    // 开启定时心跳的任务
    private void startHeartBeatTask() {
        new Thread(() -> {
            while (true) {
                try {
                    TimeUnit.MILLISECONDS.sleep(GlobalConstant.SERVICE_HEART_BEAT_TIME_STEP);

                    nameServerClient.sendMsgSync(EventCodeEnum.HEART_BEAT_REQ, new ServiceHeartBeatReq());
                } catch (InterruptedException e) {
                    log.error("", e);
                }
            }
        }, "heart-beat-task-thread").start();
    }

    // 开启定时刷新 broker 客户端连接的任务
    private void startRefreshBrokerConnectionTask() {
        new Thread(() -> {
            while (true) {
                try {
                    TimeUnit.MILLISECONDS.sleep(GlobalConstant.SERVICE_HEART_BEAT_TIME_STEP);

                    connectBroker();
                } catch (InterruptedException e) {
                    log.error("", e);
                }
            }
        }, "fetch-broker-address-task-thread").start();
    }

    // 与 broker 建立连接
    private void connectBroker() {
        // 获取 broker 的地址集合
        TcpMsg resp = nameServerClient.sendMsgSync(
                EventCodeEnum.PULL_BROKER_ADDRESS_REQ,
                new PullBrokerAddressReq(brokerRole, brokerGroup)
        );
        List<String> brokerAddressList = JSON.parseObject(resp.getBody(), PullBrokerAddressResp.class)
                .getBrokerAddressList();
        AssertUtil.isNotEmpty(brokerAddressList, "没有匹配的 broker");

        // 判断是否需要修改 brokerClientMap
        boolean changed = brokerAddressList.size() != brokerClientMap.size();
        Map<String, NettyClient> newBrokerClientMap = new ConcurrentHashMap<>(brokerAddressList.size());
        for (String brokerAddress : brokerAddressList) {
            // 如果连接过这个 broker，则无需连接
            NettyClient oldBrokerClient = brokerClientMap.get(brokerAddress);
            if (oldBrokerClient != null) {
                newBrokerClientMap.put(brokerAddress, oldBrokerClient);
                continue;
            }

            changed = true;
            NettyClient brokerClient = new NettyClient(brokerAddress);
            String clientName = genBrokerClientName(brokerAddress);
            brokerClient.initConnection(clientName, new BrokerRespHandler(new EventBus(clientName + "-event-bus")));
            newBrokerClientMap.put(brokerAddress, brokerClient);
        }

        // 如果需要修改，则在同步代码块中修改
        if (changed) {
            synchronized (this) {
                brokerClientMap = newBrokerClientMap;
            }
        }
    }

    /**
     * 根据 broker 的地址生成 broker 客户端的名称
     *
     * @param brokerAddress broker 的地址
     * @return broker 客户端的名称
     */
    protected abstract String genBrokerClientName(String brokerAddress);

    /**
     * 当客户端启动后会回调这个方法，默认空实现
     */
    protected void afterStartCallback() {
    }

    /**
     * 从 brokerClientMap 中挑选出一个 broker 客户端
     *
     * @return 一个 broker 客户端
     */
    protected NettyClient pickBrokerClient() {
        // TODO 这里使用第一个 broker 作为发送消息的 broker，之后需要灵活一点，详细内容见 7-7
        return new ArrayList<>(brokerClientMap.values()).get(0);
    }
}
